package com.fxkj.cloud.drms.infra.sql.TDengine;

import com.fxkj.cloud.dbs.common.constants.DataType;
import com.fxkj.cloud.dbs.common.constants.DeviceStatusEnum;
import com.fxkj.cloud.dbs.common.model.DeviceAttributeKey;
import com.fxkj.cloud.dbs.common.model.aggregation.AggregationType;
import com.fxkj.cloud.dbs.common.model.aggregation.QueryType;
import com.fxkj.cloud.dbs.common.model.aggregation.TsQuery;
import com.fxkj.cloud.dbs.common.model.aggregation.TsQueryResult;
import com.fxkj.cloud.dbs.common.model.kv.*;
import com.fxkj.cloud.drms.domain.device.repository.*;
import com.fxkj.cloud.drms.infra.config.TDengineJdbcTemplateConfig;
import com.fxkj.cloud.drms.infra.utils.FxExecutors;
import com.google.common.base.Function;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import org.springframework.beans.factory.annotation.Value;

import jakarta.annotation.Nullable;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

public abstract class TDEngineAbstractAsyncDao extends TDEngineAbstractDao {

    protected ExecutorService readResultsProcessingExecutor;

    @Value("${fx.ts.tdengine.query.result_processing_threads:50}")
    protected int threadPoolSize;

    public TDEngineAbstractAsyncDao(TDEngineBufferedRateReadExecutor readExecutor, TDEngineBufferedRateWriteExecutor writeExecutor, TDengineJdbcTemplateConfig tdEngineJdbcTemplateConfig) {
        super(readExecutor, writeExecutor, tdEngineJdbcTemplateConfig);
    }

    @PostConstruct
    public void startExecutor() {
        readResultsProcessingExecutor = FxExecutors.newWorkStealingPool(threadPoolSize, "mysql-callback");
    }

    @PreDestroy
    public void stopExecutor() {
        if (readResultsProcessingExecutor != null) {
            readResultsProcessingExecutor.shutdownNow();
        }
    }


    protected <T, S> ListenableFuture<S> getFuture(TDEngineResultSetFuture<T> future, java.util.function.Function<TDEngineResultSet<T>, S> transformer) {
        return Futures.transform(future, new Function<TDEngineResultSet<T>, S>() {
            @Nullable
            @Override
            public S apply(@Nullable TDEngineResultSet<T> input) {
                return transformer.apply(input);
            }
        }, readResultsProcessingExecutor);
    }

    protected <T, S> ListenableFuture<S> getFutureAsync(TDEngineResultSetFuture<T> future, AsyncFunction<TDEngineResultSet<T>, S> transformer) {
        return Futures.transformAsync(future, new AsyncFunction<TDEngineResultSet<T>, S>() {
            @Nullable
            @Override
            public ListenableFuture<S> apply(@Nullable TDEngineResultSet<T> input) {
                try {
                    return transformer.apply(input);
                } catch (Exception e) {
                    return Futures.immediateFailedFuture(e);
                }
            }
        }, readResultsProcessingExecutor);
    }

    protected ListenableFuture<List<TsKvEntry>> convertHistoryTsKvEntries(TDEngineResultSet<ResultSet> rs, Map<String, DataType> keyTypeMap) {
        return Futures.transform(rs.tskvEntriesResult(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            List<TsKvEntry> tsKvEntries = new ArrayList<>();
                            try {
                                while (null != resultSet && resultSet.next()) {
                                    long ts = resultSet.getLong(SqlConstants.TS);
                                    keyTypeMap.forEach((key, dataType) -> {
                                        tsKvEntries.add(TskvEntryConverter.convert(ts, key, dataType, resultSet));
                                    });
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return tsKvEntries;
                        }
                ),
                input -> input, this.readResultsProcessingExecutor);
    }

    protected ListenableFuture<List<TsKvEntry>> convertToLatestTsKvEntries(TDEngineResultSet<ResultSet> rs, Map<String, DataType> keyTypeMap) {
        return Futures.transform(rs.tskvEntriesResult(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            Map<String, TsKvEntry> tsKvEntryMap = new HashMap<>();
                            try {
                                while (null != resultSet && resultSet.next()) {
                                    long ts = resultSet.getLong(SqlConstants.TS);
                                    keyTypeMap.forEach((key, value) -> {
                                        TsKvEntry tsKvEntry = tsKvEntryMap.get(key);
                                        //和已有值进行判断
                                        if (null == tsKvEntry) {
                                            tsKvEntry = TskvEntryConverter.convert(ts, key, value, resultSet);
                                            tsKvEntryMap.put(key, tsKvEntry);
                                        } else {
                                            if (tsKvEntry.getTs() <= ts) {
                                                tsKvEntryMap.put(key, TskvEntryConverter.convert(ts, key, value, resultSet));
                                            }
                                        }
                                    });
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return new ArrayList<>(tsKvEntryMap.values());
                        }
                ),
                input -> input, this.readResultsProcessingExecutor);
    }

    protected ListenableFuture<List<TsKvEntry>> convertToLatestTsKvEntry(TDEngineResultSet<ResultSet> rs, List<String> attributeKeys) {
        return Futures.transform(rs.tskvEntriesResult(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            Map<String, TsKvEntry> tsKvEntryMap = new HashMap<>();
                            try {
                                while (null != resultSet && resultSet.next()) {
                                    long ts = resultSet.getLong(SqlConstants.TS);
                                    attributeKeys.forEach(key -> {
                                        TsKvEntry tsKvEntry = tsKvEntryMap.get(key);
                                        //和已有值进行判断
                                        try {
                                            if (null == tsKvEntry) {

                                                tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(key, resultSet.getString(key)));
                                                tsKvEntryMap.put(key, tsKvEntry);
                                            } else {
                                                if (tsKvEntry.getTs() <= ts) {
                                                    tsKvEntryMap.put(key, new BasicTsKvEntry(ts, new StringDataEntry(key, resultSet.getString(key))));
                                                }
                                            }
                                        } catch (SQLException e) {
                                            throw new RuntimeException(e);
                                        }
                                    });
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return new ArrayList<>(tsKvEntryMap.values());
                        }
                ),
                input -> input, this.readResultsProcessingExecutor);
    }

    protected ListenableFuture<Map<UUID, Map<String, KvEntry>>> convertToProductLatestTsKvEntry(TDEngineResultSet<ResultSet> rs, List<String> attributeKeys) {
        return Futures.transform(rs.transform(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            Map<UUID, Map<String, KvEntry>> tsKvEntryMap = new HashMap<>();
                            try {
                                while (null != resultSet && resultSet.next()) {
                                    UUID deviceId = SqlConstants.tableNameToUUID((resultSet.getString(SqlConstants.DEVICE_ID)));
                                    List<KvEntry> tsKvEntries = new ArrayList<>();
                                    for (String attributeKey : attributeKeys) {
                                        tsKvEntries.add(new StringDataEntry(attributeKey, resultSet.getString(attributeKey)));
                                    }
                                    tsKvEntryMap.put(deviceId, tsKvEntries.stream().collect(Collectors.toMap(KvEntry::getKey, kvEntry -> {
                                        //若为空则补0
                                        if (null == kvEntry.getValue()) {
                                            return new StringDataEntry(kvEntry.getKey(), "0");
                                        }
                                        return kvEntry;
                                    })));
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return tsKvEntryMap;
                        }
                ),
                input -> input, this.readResultsProcessingExecutor);

    }

    protected ListenableFuture<ProductStatusStatistics> convertToDeviceStatsStatistics(TDEngineResultSet<ResultSet> rs) {
        return Futures.transform(rs.transform(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            List<DeviceStatusStatistics> deviceStatusStatistics = new ArrayList<>();
                            try {
                                while (null != resultSet && resultSet.next()) {
                                    Integer lastStatus = resultSet.getInt(SqlConstants.LAST_STATUS);
                                    Long num = resultSet.getLong(SqlConstants.STATUS_NUM);
                                    deviceStatusStatistics.add(DeviceStatusStatistics.create(lastStatus, num));
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return deviceStatusStatistics;
                        }
                ),
                input -> ProductStatusStatistics.deviceStatusStatistics(input), this.readResultsProcessingExecutor);
    }

    protected ListenableFuture<ProductDeviceStatusChangeHistory> convertToProductDeviceStatsChangeHistory(TDEngineResultSet<ResultSet> rs) {
        return Futures.transform(rs.transform(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            List<ProductDeviceStatusChangeHistoryDetail> details = new ArrayList<>();
                            List<ProductDeviceStatusStatistics> statistics = new ArrayList<>();
                            try {
                                Long offlineCount = 0L;
                                Long onlineCount = 0L;
                                Long workingCount = 0L;
                                Long alarmCount = 0L;
                                while (null != resultSet && resultSet.next()) {
                                    Long startTs = resultSet.getLong(SqlConstants.COUNT_TIME);
                                    long total = resultSet.getLong(SqlConstants.SUM_COUNT);
                                    offlineCount = resultSet.getLong(SqlConstants.OFFLINE_COUNT);
                                    onlineCount = resultSet.getLong(SqlConstants.ONLINE_COUNT);
                                    workingCount = resultSet.getLong(SqlConstants.WORKING_COUNT);
                                    alarmCount = resultSet.getLong(SqlConstants.ALARM_COUNT);
                                    details.add(ProductDeviceStatusChangeHistoryDetail.create(startTs, offlineCount, DeviceStatusEnum.OFFLINE.getCode(), total));
                                    details.add(ProductDeviceStatusChangeHistoryDetail.create(startTs, onlineCount, DeviceStatusEnum.ONLINE.getCode(), total));
                                    details.add(ProductDeviceStatusChangeHistoryDetail.create(startTs, workingCount, DeviceStatusEnum.WORKING.getCode(), total));
                                    details.add(ProductDeviceStatusChangeHistoryDetail.create(startTs, alarmCount, DeviceStatusEnum.ALARMING.getCode(), total));
                                    statistics.add(ProductDeviceStatusStatistics.create(startTs, total, offlineCount, onlineCount, workingCount, alarmCount));
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return ProductDeviceStatusChangeHistory.create(statistics, details);
                        }
                ),
                input -> input, this.readResultsProcessingExecutor);
    }

    protected ListenableFuture<List<ProductDeviceLightStatistics>> convertToProductDeviceLightStatistics(TDEngineResultSet<ResultSet> rs) {
        return Futures.transform(rs.transform(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            List<ProductDeviceLightStatistics> productDeviceLightStatistics = new ArrayList<>();
                            try {
                                while (null != resultSet && resultSet.next()) {
                                    productDeviceLightStatistics.add(
                                            ProductDeviceLightStatistics.create(
                                                    resultSet.getLong(SqlConstants.START_TIME),
                                                    resultSet.getLong(SqlConstants.OPEN_NUM)
                                            )
                                    );
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return productDeviceLightStatistics;
                        }
                ),
                input -> input, this.readResultsProcessingExecutor);
    }

    protected ListenableFuture<List<DeviceStatusLatestInfo>> convertToDeviceStatusLatestInfo(TDEngineResultSet<ResultSet> rs) {
        return Futures.transform(rs.transform(
                        this.readResultsProcessingExecutor,
                        resultSet -> {
                            List<DeviceStatusLatestInfo> deviceStatusLatestInfos = new ArrayList<>();
                            try {
                                while (null != resultSet && resultSet.next()) {
                                    deviceStatusLatestInfos.add(
                                            DeviceStatusLatestInfo.create(
                                                    UUID.fromString(resultSet.getString(SqlConstants.DEVICE_ID)),
                                                    resultSet.getInt(SqlConstants.LAST_STATUS)
                                            )
                                    );
                                }
                            } catch (SQLException e) {
                                throw new RuntimeException(e);
                            }
                            return deviceStatusLatestInfos;
                        }
                ),
                input -> input, this.readResultsProcessingExecutor);
    }

    protected ListenableFuture<List<TsQueryResult>> convertToTsQueryResult(TDEngineResultSet<ResultSet> rs, List<String> queryKeys, Boolean intervalExisted, QueryType queryType, AggregationType aggregationType) {
        return Futures.transform(rs.transform(
                this.readResultsProcessingExecutor,
                resultSet -> {
                    List<TsQueryResult> tsQueryResults = new ArrayList<>();
                    try {
                        while (null != resultSet && resultSet.next()) {
                            Map<String, String> kvs = new HashMap<>();
                            String key;
                            for (int i = 0; i < queryKeys.size(); i++) {
                                key = queryKeys.get(i);
                                kvs.put(key, resultSet.getString(key));
                            }
                            if (intervalExisted) {
                                tsQueryResults.add(TsQueryResult.of(null, resultSet.getLong(SqlConstants.TS), kvs));
                            } else {
                                tsQueryResults.add(TsQueryResult.of(
                                        queryType.equals(QueryType.DEVICE) ? SqlConstants.StringToUUID(resultSet.getString(SqlConstants.DEVICE_ID)) : SqlConstants.tableNameToUUID(resultSet.getString(SqlConstants.DEVICE_ID)),
                                        (queryType.equals(QueryType.PRODUCT) || queryType.equals(QueryType.DEVICE_TYPE)) && !aggregationType.equals(AggregationType.NONE) ? null : resultSet.getLong(SqlConstants.TS),
                                        kvs));
                            }
                        }
                    } catch (SQLException e) {
                        throw new RuntimeException(e);
                    }
                    return tsQueryResults;
                }
        ), input -> input, this.readResultsProcessingExecutor);
    }

    protected void setParams(PreparedStatement prepStmt, int index, TsKvEntry tsKvEntry) throws SQLException {
        DataType dataType = tsKvEntry.getDataType();
        switch (dataType) {
            case STRING:
                prepStmt.setString(index, tsKvEntry.getStrValue().orElse(""));
                break;
            case LONG:
                if (tsKvEntry.getLongValue().isPresent()) {
                    prepStmt.setLong(index, tsKvEntry.getLongValue().get());
                } else {
                    prepStmt.setNull(index, Types.NULL);
                }
                break;
            case BOOLEAN:
                if (tsKvEntry.getBooleanValue().isPresent()) {
                    prepStmt.setBoolean(index, tsKvEntry.getBooleanValue().get());
                } else {
                    prepStmt.setNull(index, Types.NULL);
                }
                break;
            case DOUBLE:
                if (tsKvEntry.getDoubleValue().isPresent()) {
                    prepStmt.setDouble(index, tsKvEntry.getDoubleValue().get());
                } else {
                    prepStmt.setNull(index, Types.NULL);
                }
                break;
            case JSON:
                prepStmt.setString(index, tsKvEntry.getJsonValue().orElse(""));
                break;
        }
    }

    protected PreparedStatement getSavePrepStmt(UUID productId, List<TsKvEntry> tsKvEntries) {
        String sql = SqlConstants.getBatchInsertSql(productId, tsKvEntries);
        return prepare(sql);
    }

    protected PreparedStatement getSuperTableCreatePrepStmt(UUID rootOrganizationId, UUID productId, Map<String, DataType> attributeKeyDataTypeMap) {
        String sql = SqlConstants.getSuperTableCreateSql(rootOrganizationId, productId, attributeKeyDataTypeMap);
        return prepare(sql);
    }

    protected List<PreparedStatement> getAddProductAttributePrepStmt(UUID productId, Map<String, DataType> attributeKeyDataTypeMap) {
        List<String> sqls = SqlConstants.getAddProductAttributeSql(productId, attributeKeyDataTypeMap);
        List<PreparedStatement> preps = new ArrayList<>();
        for (String sql : sqls) {
            preps.add(prepare(sql));
        }
        return preps;
    }

    protected List<PreparedStatement> getRemoveProductAttributePrepStmt(UUID productId, Set<String> attributeKeys) {
        List<String> sqls = SqlConstants.getRemoveProductAttributeSql(productId, attributeKeys);
        List<PreparedStatement> preparedStatements = new ArrayList<>();
        return sqls.stream().map(sql -> prepare(sql)).collect(Collectors.toList());
    }

    protected PreparedStatement getSonTableCreatePrepStmt(UUID rootOrganizationId, UUID productId, UUID deviceId) throws SQLException {
        String sql = SqlConstants.getSonTableCreateSql(rootOrganizationId, productId, deviceId);
        PreparedStatement prepare = prepare(sql);
        prepare.setString(1, SqlConstants.UUIDToString(productId));
        prepare.setString(2, SqlConstants.UUIDToString(deviceId));
        return prepare;
    }

    protected PreparedStatement getSelectLatestPrepStmt(UUID deviceId, List<DeviceAttributeKey> deviceAttributeKeys) {
        String sql = SqlConstants.getSelectLatest(deviceId, deviceAttributeKeys);
        return prepare(sql);
    }

    protected PreparedStatement getKeysSelectLatestPrepStmt(UUID deviceId, List<String> attributeKeys) {
        String sql = SqlConstants.getKeySelectLatest(deviceId, attributeKeys);
        return prepare(sql);
    }

    protected PreparedStatement getProductSelectLatestPrepStmt(Set<UUID> productIds, List<String> attributeKeys) {
        String sql = SqlConstants.getProductSelectLatest(productIds, attributeKeys);
        return prepare(sql);
    }

    protected PreparedStatement getSelectHistoryPrepStmt(UUID deviceId, List<DeviceAttributeKey> deviceAttributeKeys, Integer limit) {
        String sql = SqlConstants.getSelectHistory(deviceId, deviceAttributeKeys, limit);
        return prepare(sql);
    }

    protected PreparedStatement getDeviceStatusPrepStmt(UUID deviceId) {
        String sql = SqlConstants.getDeviceStatusSaveSql(deviceId);
        return prepare(sql);
    }

    protected PreparedStatement statisticsProductLatestStatusPrepStmt(Integer productNum) {
        String sql = SqlConstants.statisticsProductLatestStatusPrepStmt(productNum);
        return prepare(sql);
    }

    protected PreparedStatement getProductDeviceStatusChangeHistoryPrepStmt(Set<UUID> productIds, UUID productId, UUID deviceId, String interval) {
        String sql = SqlConstants.getDeviceStatusChangeHistoryQuery(productIds,productId, deviceId, interval);
        return prepare(sql);
    }

    protected PreparedStatement getProductLightStatisticsPrepStmt(Set<UUID> productIds) {
        String sql = SqlConstants.getProductLightStatisticsQuery(productIds);
        return prepare(sql);
    }

    protected PreparedStatement getDeviceLatestStatusInfoPrepStmt() {
        String sql = SqlConstants.getDeviceLatestStatusInfoQuery();
        return prepare(sql);
    }

    protected PreparedStatement getTsQueryPrepStmt(TsQuery query) {
        String sql = SqlConstants.getTsQuery(query);
        return prepare(sql);
    }

    protected PreparedStatement getProductOrModelEnergyRankPrepStmt(Set<UUID> productIds, Integer limit) {
        String sql = SqlConstants.getProductOrModelEnergyRankQuery(productIds, limit);
        return prepare(sql);
    }

}
